package com.verlet.springbootdemo.simple.nettyrpc.client;

import com.verlet.springbootdemo.simple.nettyrpc.RpcInfo;
import com.verlet.springbootdemo.simple.nettyrpc.RpcResult;
import com.verlet.springbootdemo.simple.nettyrpc.client.codec.RpcInfoEncoder;
import com.verlet.springbootdemo.simple.nettyrpc.client.codec.RpcResultDecoder;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.json.JsonObjectDecoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.concurrent.DefaultEventExecutorGroup;
import io.netty.util.concurrent.EventExecutorGroup;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;

import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeoutException;

/**
 * Netty 客户端
 *
 * @author Verlet
 * @date 2023-03-07 22:57
 */
@Slf4j
public class NettyRpcClientService {

    @Value("${nettyRpcHost:127.0.0.1}")
    private String nettyRpcHost;

    @Value("${nettyRpcPort:5201}")
    private int nettyRpcPort;

    @Value("${nettyRpcTimeout:180}")
    private long nettyRpcTimeout;

    /**
     * RPC 调用结果
     */
    private static final ConcurrentHashMap<Long, RpcResult> CONCURRENT_HASH_MAP = new ConcurrentHashMap<>();

    private NioEventLoopGroup workGroup;
    private Channel channel;

    private Bootstrap bootstrap;

    /**
     * Context中添加线程池,当复用channel时候，EventExecutorGroup 会被堵塞复用。
     */
    private static final EventExecutorGroup GROUP = new DefaultEventExecutorGroup(Runtime.getRuntime().availableProcessors() * 2);

    @PostConstruct
    private void init() throws InterruptedException {
        // 使用netty 发送调用信息给服务提供方
        workGroup = new NioEventLoopGroup();
        bootstrap = new Bootstrap();

        bootstrap.group(workGroup)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new LoggingHandler(LogLevel.INFO))
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline()
                                // 解码器
                                .addLast(new JsonObjectDecoder())
                                .addLast(new StringDecoder())
                                .addLast(new RpcResultDecoder())
                                // 编码器
                                .addLast(new StringEncoder(StandardCharsets.UTF_8))
                                .addLast(new RpcInfoEncoder())
                                // 处理器
                                .addLast(GROUP, new RpcClientHandler(CONCURRENT_HASH_MAP));

                    }
                });

        channel = connect();
        log.info("NettyRpcClientService 启动成功。。。");
        // 同步等待调用信息发送成功
//            channelFuture.channel().writeAndFlush(rpcInfo).sync();
        // 同步等待rpcClientHandler的channelRead被触发后（意味着收到了调用结果）
//            channelFuture.channel().closeFuture().sync();

    }


    /**
     * 连接服务器
     */
    private Channel connect() throws InterruptedException {
        // connect 是异步的，但调用其future的sync则是同步等待连接成功
        InetSocketAddress inetSocketAddress = new InetSocketAddress(nettyRpcHost, nettyRpcPort);
        ChannelFuture channelFuture = bootstrap.connect(inetSocketAddress).sync();
        return channelFuture.channel();
    }

    /**
     * 获取活动的 channel
     */
    private Channel getChannel() throws InterruptedException {
        if (!channel.isActive()) {
            // channel 不活跃重新创建连接获取
            channel = connect();
            log.info("重新获取连接");
        }
        return channel;
    }


    public Object getRpcResult(RpcInfo rpcInfo) throws Exception {
        // 同步等待调用信息发送成功
        getChannel().writeAndFlush(rpcInfo).sync();
        RpcResult result = null;
        LocalDateTime endLocalDateTime = LocalDateTime.now().plusSeconds(nettyRpcTimeout);
        do {
            if (LocalDateTime.now().isAfter(endLocalDateTime)) {
                log.error("NettyRpc 调用超时");
                break;
            }
            Thread.yield();
            result = CONCURRENT_HASH_MAP.get(rpcInfo.getRequestId());
        } while (result == null);
        CONCURRENT_HASH_MAP.remove(rpcInfo.getRequestId());
        if(result == null){
            throw new TimeoutException("NettyRpc 调用超时");
        }
        if(result.isError()){
            throw new Exception(result.getCause());
        }
        return result.getResult();

    }


    @PreDestroy
    public void close() {
        if (!workGroup.isShutdown()) {
            workGroup.shutdownGracefully();
            GROUP.shutdownGracefully();
            CONCURRENT_HASH_MAP.clear();
        }
        log.info("NettyRpcClientService 关闭成功。。。");
    }

}
